package rocketmq.quickstart;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.MessageListener;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class FIFOMessage {
  private static final Logger logger = LoggerFactory.getLogger(FIFOMessage.class);

  public static void sendFIFOMessage() throws ClientException, IOException {
    String endpoint = "localhost:8081";
    // 消息发送的目标Topic名称，需要提前创建。
    String topic = "FIFOTopic";
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
    ClientConfiguration configuration = builder.build();
    // 初始化Producer时需要设置通信配置以及预绑定的Topic。
    Producer producer = provider.newProducerBuilder()
        .setTopics(topic)
        .setClientConfiguration(configuration)
        .build();

    // 定时/延时消息发送
    MessageBuilder messageBuilder = new MessageBuilderImpl();
    ;
    Message message = messageBuilder.setTopic("FIFOTopic")
        // 设置消息索引键，可根据关键字精确查找某条消息。
        .setKeys("messageKey")
        // 设置消息Tag，用于消费端根据指定Tag过滤消息。
        .setTag("messageTag")
        .setMessageGroup("FIFOGroup")
        // 消息体
        .setBody("messageBody".getBytes())
        .build();

    try {
      // 发送消息，需要关注发送结果，并捕获失败等异常。
      SendReceipt sendReceipt = producer.send(message);
      logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
    } catch (ClientException e) {
      logger.error("Failed to send message", e);
    }
    producer.close();
  }

  // 消费示例一：使用PushConsumer消费普通消息，只需要在消费监听器中处理即可。
  public static void pushConsumer() throws ClientException, InterruptedException, IOException {
    MessageListener messageListener = new MessageListener() {
      @Override
      public ConsumeResult consume(MessageView messageView) {
        ByteBuffer byteBuffer = messageView.getBody();
        byte[] byteArray = new byte[byteBuffer.remaining()];
        byteBuffer.get(byteArray);
        String messageBody = new String(byteArray, StandardCharsets.UTF_8);
        logger.info("pushConsumer read message successfully - messageId: {}, messageBody: {}",
            messageView.getMessageId(), messageBody);
        // 根据消费结果返回状态。
        return ConsumeResult.SUCCESS;
      }
    };

    final ClientServiceProvider provider = ClientServiceProvider.loadService();
    String endpoints = "localhost:8081";
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
        .setEndpoints(endpoints)
        .build();
    // 订阅消息的过滤规则，表示订阅所有Tag的消息。
    String tag = "*";
    FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
    // 为消费者指定所属的消费者分组，Group需要提前创建。
    String consumerGroup = "FIFOGroup";
    // 指定需要订阅哪个目标Topic，Topic需要提前创建。
    String topic = "FIFOTopic";
    PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        .setClientConfiguration(clientConfiguration)
        // 设置消费者分组。
        .setConsumerGroup(consumerGroup)
        // 设置预绑定的订阅关系。
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // 设置消费监听器。
        .setMessageListener(messageListener)
        .build();
    Thread.sleep(Long.MAX_VALUE);
    // 如果不需要再使用 PushConsumer，可关闭该实例。
    pushConsumer.close();
  }

  // 消费示例二：使用SimpleConsumer消费普通消息，主动获取消息进行消费处理并提交消费结果。
  public static void simpleConsumer() throws ClientException {
    final ClientServiceProvider provider = ClientServiceProvider.loadService();
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
        .setEndpoints("localhost:8081") // NameServer 地址
        .build();

    SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
        .setClientConfiguration(clientConfiguration)
        .setConsumerGroup("FIFOGroup")
        .setSubscriptionExpressions(Collections.singletonMap("FIFOTopic",
            FilterExpression.SUB_ALL))
        .setAwaitDuration(Duration.ofSeconds(10))
        .build();

    List<MessageView> messageViewList = null;
    try {
      messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
      messageViewList.forEach(messageView -> {
        ByteBuffer byteBuffer = messageView.getBody();
        byte[] byteArray = new byte[byteBuffer.remaining()];
        byteBuffer.get(byteArray);
        String messageBody = new String(byteArray, StandardCharsets.UTF_8);
        logger.info("simpleConsumer read message successfully - messageId: {}, messageBody: {}",
            messageView.getMessageId(), messageBody);
        // 消费处理完成后，需要主动调用ACK提交消费结果。
        try {
          simpleConsumer.ack(messageView);
        } catch (ClientException e) {
          e.printStackTrace();
        }
      });
    } catch (ClientException e) {
      // 如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求。
      e.printStackTrace();
    }
  }

  public static void main(String[] args) throws ClientException, IOException, InterruptedException {
    // sendFIFOMessage(); // 发送消息到topic
    // pushConsumer(); // 接收消息从topic
    // simpleConsumer(); // 接收消息从topic
  }
}
